package org.example

import org.apache.spark.sql.{DataFrame, SparkSession}

object DataFromMysql {
  def main(args: Array[String]): Unit = {
    //创建 sparkSession 对象
    val sparkSession: SparkSession = SparkSession.builder()
      .master(master = "local[4]")
      .appName(name = "DataFromMysql")
      .getOrCreate()
    // 读取sql数据需要提前设置 url, table 和 properties
    val data: DataFrame = sparkSession.read
      .format(source = "jdbc")
      .option("url", "jdbc:mysql://localhost:3306/city")
      .option("dbtable", "city")
      .option("user", "root")
      .option("password", "root")
      .load()
    println("=================获取数据=====================")
    data.show()
    data.printSchema()

    println("====================获取前三行数据=========================")
    var myDate = data.limit(3)
    myDate.show()

    println("===================获取前四条数据===========================")
    data.show(4)

    println("=====================获取数组=================================")
    var arr = data.collect()
    println("arr(0):" + arr(0) + "\n" + "arr(1):" + arr(1))

    println("==================获取数组前6条======================")
    var arr1 = data.takeAsList(6)
    arr1.forEach(x => println(x))

    println("=====================获取符合条件的所有数据=============================")
    data.where("name='Kabul' or Population = 237500").show()
    data.printSchema()

    println("==========================指定显示列====================================")
    data.select(data("District"), data("Population") + 1).show()
    data.printSchema()

    println("===============================使用函数式的添加==============================")
    data.selectExpr("ID", "Name", "round((Population/100),2)").show()
    data.printSchema()

    println("===============================指定列，默认升序==================================")
    data.orderBy("Population").show()
    data.printSchema()

    println("========================多个字段加上某值============================")
    data.select(data.col("Name"), data.col("Population").plus(1)).show()
    data.printSchema()

    println("==========================过滤值====================================")
    data.filter(data.col("Population").gt(559000)).show()
    data.printSchema()

    println("==========================分组=======================================")
    data.groupBy("CountryCode").count().show()
    data.printSchema()

    data.select("CountryCode").distinct().show()
    data.printSchema()

    data.groupBy("CountryCode").mean().show()
    data.printSchema()

    data.groupBy("CountryCode", "District").count().show()
    data.printSchema()
  }
}
